package com.beiding.jsonanalyzer.analyze;

import java.util.function.Consumer;

/**
 * 计数窗口
 */
public class CountWindow implements Window {

    private int count;

    public CountWindow(int count) {
        if (count <= 0) {
            throw new IllegalArgumentException("计数窗口数量必须大于0");
        }
        this.count = count;
    }

    public static CountWindow of(int num) {
        return new CountWindow(num);
    }

    @Override
    public WindowInstance createInstance(Consumer<Object> open, Consumer<Object> close) {
        return new C(open, close);
    }

    private class C implements WindowInstance {

        Consumer<Object> open;
        Consumer<Object> close;

        public C(Consumer<Object> open, Consumer<Object> close) {
            this.open = open;
            this.close = close;
        }

        /**
         * 线程不安全
         * 需要在外部自行确保线程安全
         */
        int c = 0;

        Long batch = 0L;

        @Override
        public void add(Runnable action) {
            action.run();
            c++;
            if (c == count) {
                close.accept(batch);
                batch++;
                open.accept(batch);
                //重置
                c = 0;
            }

        }

        @Override
        public void start() {
            open.accept(batch);
        }

        @Override
        public void end() {
            close.accept(batch);
            //结束后进入新的批
            batch++;
        }
    }

}
